package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import com.atguigu.tingshu.album.mapper.*;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.album.service.AliyunGreenService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private AliyunGreenService aliyunGreenService;

    @Autowired
    private KafkaService kafkaService;

    /**
     * 当前用户保存专辑
     *
     * @param userId      用户ID
     * @param albumInfoVo 专辑信息VO对象
     */
    @Override
    @Transactional(rollbackFor = Exception.class)  //默认事务回滚异常：RuntimeException
    public void saveAlbumInfo(Long userId, AlbumInfoVo albumInfoVo) {
        //1.保存专辑信息
        //1.1 将前端提交的专辑信息VO对象拷贝为PO对象-同名同类型属性直接拷贝即可
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);

        //1.2 手动未其他属性赋值
        albumInfo.setUserId(userId);
        albumInfo.setTracksForFree(5);
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_NO_PASS);

        //1.3 执行保存专辑，得到专辑ID
        albumInfoMapper.insert(albumInfo);
        Long albumInfoId = albumInfo.getId();

        //2.保存专辑标签
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
                //2.1 将专辑统计VO转为PO对象
                AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                //2.2 关联专辑ID
                albumAttributeValue.setAlbumId(albumInfoId);
                //2.3 保存专辑标签关系
                albumAttributeValueMapper.insert(albumAttributeValue);
            }
        }

        //3.初始化专辑统计信息
        this.saveAlbumStat(albumInfoId, SystemConstant.ALBUM_STAT_PLAY, 0);
        this.saveAlbumStat(albumInfoId, SystemConstant.ALBUM_STAT_SUBSCRIBE, 0);
        this.saveAlbumStat(albumInfoId, SystemConstant.ALBUM_STAT_BUY, 0);
        this.saveAlbumStat(albumInfoId, SystemConstant.ALBUM_STAT_COMMENT, 0);

        //4.对内容（专辑标题，简介）进行文本审核
        try {
            Map map = aliyunGreenService.textReview(albumInfo.getAlbumTitle() + albumInfo.getAlbumIntro());
            if ("pass".equals(map.get("suggestion"))) {
                //修改专辑审核状态
                albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
                albumInfoMapper.updateById(albumInfo);
                //5.发送kafka消息到上架话题中 消息内容：专辑ID
                kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfoId.toString());
            }
        } catch (Exception e) {
            log.error("[专辑服务]内容审核异常：{}", e);
        }
    }

    /**
     * 保存专家统计类型
     *
     * @param albumInfoId 专辑ID
     * @param statType    统计类型
     * @param num         数量
     */
    @Override
    public void saveAlbumStat(Long albumInfoId, String statType, int num) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumInfoId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(num);
        albumStatMapper.insert(albumStat);
    }


    /**
     * 条件分页查询当前用户发布专辑列表
     *
     * @param albumInfoQuery 查询条件对象
     * @param pageInfo       MP分页对象
     * @return
     */
    @Override
    public Page<AlbumListVo> getUserAlbumPage(AlbumInfoQuery albumInfoQuery, Page<AlbumListVo> pageInfo) {
        return albumInfoMapper.getUserAlbumPage(pageInfo, albumInfoQuery);
    }

    @Autowired
    private TrackInfoMapper trackInfoMapper;

    /**
     * 根据专辑ID删除专辑
     *
     * @param id
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfo(Long id) {
        //1.根据专辑ID查询专辑下是否关联声音 select count(*) from track_info where album_id = ? limit 1
        LambdaQueryWrapper<TrackInfo> trackInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
        trackInfoLambdaQueryWrapper.eq(TrackInfo::getAlbumId, id);
        trackInfoLambdaQueryWrapper.last("limit 1");
        Long count = trackInfoMapper.selectCount(trackInfoLambdaQueryWrapper);
        if (count > 0) {
            throw new GuiguException(500, "专辑下存在声音！");
        }
        //2.根据主键专辑ID删除专辑信息
        albumInfoMapper.deleteById(id);

        //3.根据专辑ID删除标签列表
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

        //4.根据专辑ID删除统计列表
        LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
        albumStatMapper.delete(albumStatLambdaQueryWrapper);


        //5.发送kafka 消息删除索引库专辑
        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());

    }


    @Autowired
    private RedissonClient redissonClient;

    /**
     * 根据专辑ID查询专辑信息（包含标签列表）
     * 采用Redisson分布式锁
     *
     * @param id
     * @return
     */
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从分布式缓存Redis中获取业务数据,命中缓存直接返回，未命中缓存尝试获取分布式锁
            //1.1 构建业务数据Key 形式=前缀+数据标识ID
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 从缓存中获取业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存，直接返回结果
            if (albumInfo != null) {
                return albumInfo;
            }

            //2.尝试获取分布式锁，获取锁成功执行执行查库逻辑,业务数据放入缓存,释放锁；反之进行自旋
            //2.1 构建当前业务ID对应锁Key 形式=前缀+数据标识ID:锁后缀
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 创建锁对象
            RLock lock = redissonClient.getLock(lockKey);
            //2.3 获取分布式锁
            lock.lock();
            //3.业务逻辑，查询业务执行后释放锁
            try {
                //3.1 业务执行
                albumInfo = this.getAlbumInfoFromDB(id);
                int ttl = RandomUtil.randomInt(100, 1000);
                redisTemplate.opsForValue().set(dataKey, albumInfo, RedisConstant.ALBUM_TIMEOUT + ttl, TimeUnit.SECONDS);
                return albumInfo;
            } finally {
                //3.2 释放锁采用lua脚本释放
               lock.unlock();
            }
        } catch (Exception e) {
            //缓存服务有问题，兜底处理方法
            log.error("[专辑服务]查询专辑信息异常：{}", e);
            return this.getAlbumInfoFromDB(id);
        }
    }

    /**
     * 根据专辑ID查询专辑信息（包含标签列表）
     * 采用自定义分布式锁 set k v ex nx
     *
     * @param id
     * @return
     *//*
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从分布式缓存Redis中获取业务数据,命中缓存直接返回，未命中缓存尝试获取分布式锁
            //1.1 构建业务数据Key 形式=前缀+数据标识ID
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 从缓存中获取业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存，直接返回结果
            if (albumInfo != null) {
                return albumInfo;
            }

            //2.尝试获取分布式锁，获取锁成功执行执行查库逻辑,业务数据放入缓存,释放锁；反之进行自旋
            //2.1 构建当前业务ID对应锁Key 形式=前缀+数据标识ID:锁后缀
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 构建当前线程锁的标识
            String lockVal = IdUtil.randomUUID();
            //2.2 尝试获取分布式锁利用set k v ex nx
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockVal, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
            if (flag) {
                //3.业务逻辑，查询业务执行后释放锁
                try {
                    //3.1 业务执行
                    albumInfo = this.getAlbumInfoFromDB(id);
                    int ttl = RandomUtil.randomInt(100, 1000);
                    redisTemplate.opsForValue().set(dataKey, albumInfo, RedisConstant.ALBUM_TIMEOUT + ttl, TimeUnit.SECONDS);
                    return albumInfo;
                } finally {
                    //3.2 释放锁采用lua脚本释放
                    String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                            "then\n" +
                            "    return redis.call(\"del\",KEYS[1])\n" +
                            "else\n" +
                            "    return 0\n" +
                            "end";
                    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                    redisScript.setScriptText(script);
                    redisScript.setResultType(Long.class);
                    redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockVal);
                }
            } else {
                //4.获取锁失败，自旋
                Thread.sleep(100);  //sleep睡眠,让出CPU
                this.getAlbumInfo(id);
            }
        } catch (Exception e) {
            //缓存服务有问题，兜底处理方法
            log.error("[专辑服务]查询专辑信息异常：{}", e);
            return this.getAlbumInfoFromDB(id);
        }
        return null;
    }*/


    /**
     * 从数据库中查询专辑信息
     *
     * @param id
     * @return
     */
    @Override
    @GuiGuCache(prefix = "album:info:")
    public AlbumInfo getAlbumInfoFromDB(Long id) {
        //1.根据主键辑ID查询专辑信息
        //AlbumInfo albumInfo = this.getById(id);
        AlbumInfo albumInfo = albumInfoMapper.selectById(id);
        if (albumInfo != null) {
            //2.根据专辑ID查询专辑标签列表
            LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
            albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(albumAttributeValueLambdaQueryWrapper);
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        return albumInfo;
    }

    /**
     * 修改专辑信息
     *
     * @param albumInfo
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(AlbumInfo albumInfo) {
        //1.修改专辑表基本信息
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_NO_PASS);
        albumInfoMapper.updateById(albumInfo);

        //2.修改专辑标签关系记录
        List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
        //2.1 删除该专辑下旧关联的标签记录
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, albumInfo.getId());
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);
        //2.2 新增专辑标签记录
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            for (AlbumAttributeValue albumAttributeValue : albumAttributeValueVoList) {
                //2.2 关联专辑ID
                albumAttributeValue.setAlbumId(albumInfo.getId());
                //2.3 保存专辑标签关系
                albumAttributeValueMapper.insert(albumAttributeValue);
            }
        }
        //修改后专辑内容再次进行内容审核，审核通过后才进行发布
        try {
            Map map = aliyunGreenService.textReview(albumInfo.getAlbumTitle() + albumInfo.getAlbumIntro());
            if ("pass".equals(map.get("suggestion"))) {
                //修改专辑审核状态
                albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
                albumInfoMapper.updateById(albumInfo);
                //5.发送kafka消息到上架话题中 消息内容：专辑ID
                kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfo.getId().toString());
            } else {
                kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumInfo.getId().toString());
            }
        } catch (Exception e) {
            log.error("[专辑服务]内容审核异常：{}", e);
        }
    }

    /**
     * 查询当前用户发布专辑列表
     *
     * @param userId
     * @return
     */
    @Override
    public List<AlbumInfo> getUserAllAlbumList(Long userId) {
        //1.构建查询条件对象
        LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(AlbumInfo::getUserId, userId);
        queryWrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle);
        queryWrapper.last("limit 200");
        queryWrapper.orderByDesc(AlbumInfo::getId);
        //2.执行查询
        return albumInfoMapper.selectList(queryWrapper);
    }

    /**
     * 根据专辑ID查询专辑统计信息
     *
     * @param albumId
     * @return
     */
    @Override
    @GuiGuCache(prefix = "album:info:stat:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        return albumInfoMapper.getAlbumStatVo(albumId);
    }


    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 更新统计信息
     * （幂等性处理方案）
     *
     * @param mqVo
     */
    @Override
    public void updateStata(TrackStatMqVo mqVo) {
        //1.业务要求同一更新统计MQ消息只能消费一次
        String key = RedisConstant.BUSINESS_PREFIX + "album:" + mqVo.getBusinessNo();
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, null, 10, TimeUnit.MINUTES);
        if (flag) {
            //2.首次处理业务消息
            try {
                //2.1 修改声音统计信息
                if (SystemConstant.TRACK_STAT_PLAY_NAME.equals(mqVo.getStatType())) {
                    //2.2 修改声音所属专辑统计信息
                    //更新 声音及专辑 播放量
                    trackInfoMapper.updateStat(mqVo.getTrackId(), SystemConstant.TRACK_STAT_PLAY, mqVo.getCount());
                    albumInfoMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_PLAY, mqVo.getCount());
                } else if (SystemConstant.TRACK_STAT_COMMENT_NAME.equals(mqVo.getStatType())) {
                    trackInfoMapper.updateStat(mqVo.getTrackId(), SystemConstant.TRACK_STAT_COMMENT, mqVo.getCount());
                    albumInfoMapper.updateStat(mqVo.getAlbumId(), SystemConstant.ALBUM_STAT_COMMENT, mqVo.getCount());
                }
                //TODO 其他统计类型：声音收藏、点赞
            } catch (Exception e) {
                //如果业务逻辑更新统计数据库异常，将key删除，默认情况：Kakfa会再次投递消息（默认10次未设置重试话题）
                redisTemplate.delete(key);
                throw new RuntimeException(e);
            }
        }
    }
}
